package ru.unlocker.reports.kafka;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import java.io.IOException;
import java.io.Serial;
import java.nio.charset.StandardCharsets;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.val;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import ru.unlocker.reports.domain.ReportAndTimerEvent;

@RequiredArgsConstructor
public class ReportAndTimerEventSchema implements KafkaRecordDeserializationSchema<ReportAndTimerEvent>,
    KafkaRecordSerializationSchema<ReportAndTimerEvent> {

  @Serial
  private static final long serialVersionUID = -2000138517793953864L;
  private static final ObjectMapper objectMapper = (new ObjectMapper())
      .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
      .registerModule(new JavaTimeModule());
  @NonNull
  private final String topicName;

  @Override
  public TypeInformation<ReportAndTimerEvent> getProducedType() {
    return TypeInformation.of(ReportAndTimerEvent.class);
  }

  @Override
  @SneakyThrows
  public ProducerRecord<byte[], byte[]> serialize(
      ReportAndTimerEvent element, KafkaSinkContext context, Long timestamp
  ) {
    return new ProducerRecord<>(
        topicName,
        element.getKey().getBytes(StandardCharsets.UTF_8),
        objectMapper.writeValueAsBytes(element)
    );
  }

  @Override
  public void deserialize(ConsumerRecord<byte[], byte[]> record, Collector<ReportAndTimerEvent> out)
      throws IOException {
    val element = objectMapper.readValue(record.value(), ReportAndTimerEvent.class);
    out.collect(element);
  }
}
